package com.mylater.nace.server;

import com.mylater.nace.common.ClassExecuter;
import com.mylater.nace.common.NaceThreadFactory;
import org.jboss.netty.bootstrap.Bootstrap;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.*;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * The HotSwapServer represents
 *
 * @author fengjiachun
 * @version $Id$
 */
public class HotSwapServer extends AbstractNettyServer {

    private static final Logger LOG = LoggerFactory.getLogger(HotSwapServer.class);

    static {
        try {
            new HotSwapServer().startServer();
        } catch (Exception e) {
            LOG.error("start HotSwapServer error {}", e);
        }
    }

    @Override
    public Bootstrap createServerBootstrap() {
        ExecutorService boss = Executors.newSingleThreadExecutor(new NaceThreadFactory("HotSwapServer_Boss"));
        ExecutorService worker = Executors.newSingleThreadExecutor(new NaceThreadFactory("HotSwapServer_Worker"));
        this.serverBootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(boss, worker, 1));
        return this.serverBootstrap;
    }

    @Override
    public TRANSMISSION_PROTOCOL getTransmissionProtocol() {
        return TRANSMISSION_PROTOCOL.TCP;
    }

    @Override
    public void startServer() throws Exception {
        startServer(20001);
    }

    @Override
    public void startServer(int port) throws Exception {
        InetSocketAddress socketAddress = new InetSocketAddress(port);
        startServer(socketAddress);
    }

    @Override
    public void startServer(InetSocketAddress socketAddress) throws Exception {
        setPipelineFactory(new HotSwapPipelineFactory());
        String[] optionsList = new String[2];
        optionsList[0] = "child.tcpNoDelay";        // 关闭Nagle算法
        optionsList[1] = "child.keepAlive";         // TCP定期发送心跳包，应用层不应该依赖这个选项，应用层有自己的心跳机制
        configureServerBootStrap(optionsList);
        try {
            ((ServerBootstrap) this.serverBootstrap).bind(socketAddress);
            if (LOG.isInfoEnabled()) {
                LOG.info("HotSwapServer start, port=" + socketAddress.getPort());
            }
        } catch (ChannelException e) {
            LOG.error("Unable to start HotSwapServer due to error {}", e);
            throw e;
        }
    }

    class HotSwapPipelineFactory implements ChannelPipelineFactory {

        private SimpleChannelHandler messageReceivedHandler = new SimpleChannelHandler() {

            @Override
            public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
                byte[] classByte = (byte[]) e.getMessage();
                String resultMsg = ClassExecuter.execute(classByte);
                byte[] resultByte = resultMsg.getBytes(Charset.forName("UTF-8"));
                ChannelBuffer buffer = ChannelBuffers.buffer(resultByte.length);
                buffer.writeBytes(resultByte);
                e.getChannel().write(buffer);
            }

            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
                LOG.error("HotSwap Exception Caught: {}. Going to close channel.", e.getCause());
                e.getChannel().close();
            }

            @Override
            public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
                if (LOG.isInfoEnabled()) {
                    LOG.info("HotSwap Connected Channel with ip: {}.", e.getChannel().getRemoteAddress());
                }
            }

            @Override
            public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
                if (LOG.isInfoEnabled()) {
                    LOG.info("HotSwap Disconnected Channel with ip: {}.", e.getChannel().getRemoteAddress());
                }
            }
        };

        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return addHandlers(Channels.pipeline());
        }

        public ChannelPipeline addHandlers(ChannelPipeline pipeline) {
            if (null == pipeline) {
                return null;
            }
            pipeline.addLast("hotSwapDecoder", new FrameDecoder() {

                @Override
                protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
                    if (buffer.readableBytes() >= 4) {
                        buffer.markReaderIndex();               // 标记ReaderIndex
                        int msgBodyLen = buffer.readInt();      // 前四个字节存放消息的长度
                        if (buffer.readableBytes() >= msgBodyLen) {
                            ChannelBuffer dst = ChannelBuffers.buffer(msgBodyLen);
                            buffer.readBytes(dst, msgBodyLen);
                            return dst.array();
                        } else {
                            buffer.resetReaderIndex();
                            return null;
                        }
                    }
                    return null;
                }
            });
            pipeline.addLast("hotSwapHandler", messageReceivedHandler);
            return pipeline;
        }
    }
}
